package com.ljc.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class RabbitMQConfig {
    //交换机名称
    public static final String SECKILL_EXCHANGE = "seckill_exchange";
    public static final String ERROR_EXCHANGE = "error_exchange";
    //队列名称
    public static final String SECKILL_QUEUE = "seckill.queue";
    public static final String ERROR_QUEUE = "error.queue";
    // 绑定名称
    public static final String SECKILL_BINDING = "seckill.binding";
    public static final String ERROR_BINDING = "error.binding";
    // 路由键
    public static final String ERROR_KEY = "error";

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) throws BeansException {
        // 获取RabbitTemplate
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        //设置Json转换器
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        // 设置ReturnCallback
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            // 判断是否是延迟消息
            if (message.getMessageProperties().getReceivedDelay() > 0) {
                // 是一个延迟，互连错误提示
                return;
            }
            // 投递失败，记录日志
            log.info("消息发送失败，应答码{}，原因{}，交换机{}，路由键{},消息{}",
                    replyCode, replyText, exchange, routingKey, message);
            // 如果有业务需要，可以重发消息
        });
        return rabbitTemplate;
    }

    /**
     * Json转换器
     */
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean(name = SECKILL_QUEUE)
    public Queue confirmQueue() {
        return QueueBuilder.durable(SECKILL_QUEUE).build();
    }

    @Bean(name = ERROR_QUEUE)
    public Queue errorQueue() {
        return QueueBuilder.durable(ERROR_QUEUE).build();
    }

    /**
     * 创建一个交换机
     */
    @Bean(name = SECKILL_EXCHANGE)
    public Exchange confirmExchange() {
        return ExchangeBuilder.topicExchange(SECKILL_EXCHANGE).durable(true).build();
    }

    @Bean(name = SECKILL_BINDING)
    public Binding confirmExchange(@Qualifier(SECKILL_EXCHANGE) Exchange exchange, @Qualifier(SECKILL_QUEUE) Queue queue) {
        //bind(queue) 绑定队列 to(exchange) 绑定交换机 with("") routingKey这里绑定的是空白可以按照自己的要求绑定  noargs() 没有参数
        return BindingBuilder.bind(queue).to(exchange).with("seckill.#").noargs();
    }

    /**
     * 创建错误信息交换机
     */
    @Bean(name = ERROR_EXCHANGE)
    public Exchange errorMessageExchange() {
        return ExchangeBuilder.directExchange(ERROR_EXCHANGE).durable(true).build();
    }

    @Bean(name = ERROR_BINDING)
    public Binding errorBinding() {
        return BindingBuilder.bind(errorQueue()).to(errorMessageExchange()).with(ERROR_KEY).noargs();
    }

    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
        return new RepublishMessageRecoverer(rabbitTemplate, ERROR_EXCHANGE, ERROR_KEY);
    }
}